package job

import (
	npb "enotify/api/notify"
	"enotify/internal/xys/notify/event"
	"enotify/pkg/async/conf"
	pkgNotify "enotify/pkg/notify"
	pkgConf "enotify/pkg/notify/conf"
	"fmt"
	cluster "github.com/bsm/sarama-cluster"
	"github.com/golang/protobuf/proto"
	//"gopkg.in/Shopify/sarama.v1"
	"strings"
	"sync"
	"time"
)

type Job struct {
	c        *conf.Config
	consumer *cluster.Consumer
	worker   int
	wg sync.WaitGroup
	stop chan struct{}
	event event.Event
	senderConfig *pkgConf.Config
	sender *pkgNotify.Sender
}
func (j *Job) GetAccessToken(appId string) (string,error)  {
	return "xxxxxxx",nil
}
func (j *Job) Id() (int64,error)  {
	return time.Now().Unix(),nil
}
func NewJob(c *conf.Config,senderConf *pkgConf.Config, worker int) *Job {

	j := &Job{
		c:        c,
		//consumer: newKafkaSub(c.Kafka),
		worker:   worker,
		wg:       sync.WaitGroup{},
		stop:     make(chan struct{},0),
		event: event.NewEvent(),
		senderConfig: senderConf,
	}
	app := pkgNotify.SenderWithApp(senderConf)
	sms := pkgNotify.SenderWithSMS(senderConf,j)
	wx := pkgNotify.SenderWithWxTemplate(senderConf,j)
	sender := pkgNotify.NewSender(senderConf,app,sms,wx)
	j.sender = sender
	return j
}
func (j *Job) Start() {
	var worker = j.worker
	if worker < 1 {
		worker = 5
	}
	j.wg.Add(worker)
	fmt.Println("start work----",worker)
	for i:=0;i<worker;i++ {
		go j.consume(i)
	}
}
func (j *Job) Close() (err error) {
	if j.consumer != nil {
		err = j.consumer.Close()
	}
	close(j.stop)
	j.wg.Wait()
	return
}

func (j *Job) consume(workId int) {
	for {
		select {
		case <-j.stop:
			j.wg.Done()
			return
		case err := <-j.consumer.Errors():
			fmt.Printf("err consumer error(%v)", err)
		case n := <-j.consumer.Notifications():
			fmt.Printf("notify consumer error(%v)", n)
		case msg, ok := <-j.consumer.Messages():
			if !ok {
				j.wg.Done()
				return
			}
			j.consumer.MarkOffset(msg,"")
			sendMsg := new(npb.SendMsg)
			err := proto.Unmarshal(msg.Value,sendMsg)
			if err != nil {
				fmt.Printf("proto.Unmarshal(%v) error(%v)",msg,err)
			}
		}
	}
}
func (j *Job) push(cleintInfo npb.SendMsg)  {
	data,err := j.event.GetContentBySceneId(cleintInfo.SceneId)
	if err != nil {
		fmt.Println(err)
		return
	}
	mapString,err := j.event.TransferMapStringString(data)
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Println(mapString)
	for _,vals := range mapString{
		switch vals["msgType"] {
		case "1":
			j.sendWxTemplate(cleintInfo,vals,vals["appName"])
		case "2":
			if vals["sendType"] == "1" {
				j.sendApp(cleintInfo,vals,"xysjiguang")
				j.sendApp(cleintInfo,vals,"yxjiguang")
				continue
			}
			arr := strings.Split(vals["sendType"], ",")
			for _,v := range arr {
				if v == "2"{
					j.sendApp(cleintInfo,vals,"xysjiguang")
				} else if v == "3" {
					j.sendApp(cleintInfo,vals,"yxjiguang")
				}
			}
		case "3":
			j.sendSMS(cleintInfo,vals,"chuanglan")
		}
	}
}
func (j *Job) sendApp(msg npb.SendMsg,data map[string]string,bizType string){
	appMsg := pkgNotify.NewAppMsg(msg.Client["uid"],data["title"],data["msg"],data["url"])
	err := j.sender.SendApp(bizType,appMsg)
	fmt.Println("job app---------",err)
}
func (j *Job) sendSMS(msg npb.SendMsg,data map[string]string,bizType string){
	smsMsg := pkgNotify.NewSmsMsg(msg.Client["realMobile"],data["msg"],data["signatureId"])
	err := j.sender.SendSms(bizType,smsMsg)
	fmt.Println("job sms---------",err)
}
func (j *Job) sendWxTemplate(msg npb.SendMsg,data map[string]string,bizType string){
	templateId,err:= j.event.GetWxTemplateId(data["signatureId"])
	if err != nil {
		return
	}
	keyVals := j.event.GetWxTemplateKeyVal(data)
    wxKV := make(map[string]pkgNotify.Keyworkd,0)
    var color string
    var ok bool
	for k,v := range keyVals{
		if strings.Contains(k, "_color") {
			continue
		}
		ckey := k+"_color"
		if color,ok = keyVals[ckey]; !ok {
			color = "msg"
		}
		wxKV[k] = pkgNotify.NewKeyword(v,color)
	}
	fmt.Printf("wxKv: %+v \n",wxKV)
	template := pkgNotify.NewTemplateMsg(msg.Client["account"],templateId,data["url"],wxKV)
	fmt.Printf("template: %+v \n",template)
	err = j.sender.SendWxTemplate(bizType,template)
	fmt.Println("job wx--------",err)
}
//func newKafkaSub(c *conf.Kafka) *cluster.Consumer {
//	config := cluster.NewConfig()
//	config.Consumer.Return.Errors = true
//	config.Consumer.Offsets.Initial = sarama.OffsetNewest
//	//config.Group.Return.Notifications = true
//	consumer, err := cluster.NewConsumer(c.Brokers, "notify", []string{"test"}, config)
//	if err != nil {
//		panic(err)
//	}
//	return consumer
//}
